#include "core/async_worker.hpp"
#include "logger_api.hpp" 
#include <iostream>
#include <algorithm>
#define DEBUG



AsyncWorker::AsyncWorker(moodycamel::ConcurrentQueue<LogEntry>& main_queue,
                         BlockingQueue<LogEntry>& fallback_queue,
                         std::shared_ptr<IFormatter> formatter,
                         std::chrono::milliseconds flush_interval,
                         float mainQueueSpaceThreshold,
                         size_t fallbackBatchSize)
    : m_mainQueue(main_queue)
    , m_fallbackQueue(fallback_queue)
    , m_formatter(formatter)
    , m_running(true)
    , m_flushInterval(flush_interval)
    , m_mainQueueSpaceThreshold(mainQueueSpaceThreshold)
    , m_fallbackBatchSize(fallbackBatchSize) {
    
    if (!m_formatter) {
        std::cerr << "Error: AsyncWorker initialized with null formatter." << std::endl;
        m_running = false;
        return;
    }    

    // 启动异步工作线程
    m_workerThread = std::thread(&AsyncWorker::worker_main, this);
}

AsyncWorker::~AsyncWorker() {
    m_running = false;  // 停止工作线程

    // 唤醒可能在等待的工作线程
    m_worker_cv.notify_all();
    if (m_workerThread.joinable()) {
        m_workerThread.join(); // 等待工作线程结束
    }
    // 工作线程会在退出前自行处理剩余日志
    std::cout << "AsyncWorker thread stopped and flushed." << std::endl;
}

// 提交日志条目到队列
bool AsyncWorker::enqueue(LogEntry&& entry) {
    if (!m_running) {
        #ifdef DEBUG 
        std::cerr << "Error: AsyncWorker is not running, cannot enqueue log." << std::endl;
        #endif
        return false;
    }
    // try_enqueue是无锁的，如果队列满了会返回false
    return m_mainQueue.try_enqueue(std::move(entry));
}


void AsyncWorker::addSink(std::shared_ptr<ISink>  sink) {
    if  (!sink) return;
    std::lock_guard<std::mutex> lock(m_sinks_mutex);
    m_sinks.push_back(std::move(sink));
}


void AsyncWorker::flush() {
    const size_t BATCH_CAPACITY = LoggerConstants::DEFAULT_BATCH_CAPACITY;
    std::vector<LogEntry> batch_buffer;
    batch_buffer.resize(BATCH_CAPACITY);


    size_t iterations = 0;
    const size_t MAX_ITERATIONS = LoggerConstants::MAX_ITERATIONS; // 防止无限循环

    size_t dequeued_count;

    static size_t total_processed = 0;

    // 这是一个阻塞操作，会一直循环直到两个队列都为空
    while ((m_mainQueue.size_approx() > 0 || !m_fallbackQueue.empty() || m_pendingFallbackEntry.has_value()) && iterations < MAX_ITERATIONS) {
        ++iterations;
        // 如果有待处理的回流日志，优先处理
        if (m_pendingFallbackEntry.has_value()) {
            if (m_mainQueue.try_enqueue(std::move(m_pendingFallbackEntry.value()))) {
                m_pendingFallbackEntry.reset();
            }
        }


        // 尝试从主队列处理一批
        size_t dequeued_count = m_mainQueue.try_dequeue_bulk(batch_buffer.data(), batch_buffer.capacity());
        if (dequeued_count > 0) {
            // 这里获取m_sinks的临时副本，然后在锁外遍历
            std::vector<std::shared_ptr<ISink>> current_sinks;
            {
                std::lock_guard<std::mutex> lock(m_sinks_mutex);
                current_sinks = m_sinks;
            }


            for (size_t i = 0; i < dequeued_count; ++i) {
                auto& log_entry = batch_buffer[i];
                std::string formatted_message = m_formatter->format(log_entry);
                for (const auto& sink : current_sinks) {
                    if (sink) {
                        sink->log(log_entry.level, formatted_message);
                    }
                }
            }
        }

        // 尝试回流并处理备用队列
        if (!m_pendingFallbackEntry.has_value() && !m_fallbackQueue.empty()) {
            drain_fallback_to_main();
        }

        // 只有真的没有工作时才，yield cpu
        if (dequeued_count == 0 && m_fallbackQueue.empty() && !m_pendingFallbackEntry.has_value()) {
            std::this_thread::yield();
        }

    }
    // 循环结束后，所有队列都为空
    
    total_processed += dequeued_count;
    if (total_processed % 10000 == 0) {
        std::cerr << "Flush progress: " << total_processed
                  << " logs processed, queue size: " << m_mainQueue.size_approx() << std::endl;
    }

    // 如果是超出迭代上限退出，输出错误信息
    if (iterations >= MAX_ITERATIONS) {
        std::cerr << "Warning: AsyncWorker::flush() reached max iterations. Some logs may be lost." << std::endl;
    }
}


void AsyncWorker::notify_consumer() {
    m_worker_cv.notify_one();
}

// 异步线程的主循环
void AsyncWorker::worker_main() {
    LogEntry entry;
    auto last_flush_time = std::chrono::steady_clock::now();
    // 预分配批处理缓冲区，预设10000条日志
    // 确保这个大小不要超过队列的容量，也不要太大导致内存压力
    const size_t BATCH_CAPACITY = 10000;
    std::vector<LogEntry> batch_buffer; // 批处理 缓冲区，避免频繁调用Sink::log
    // 使用 resize() 而不是 reserve()，这会默认构造 BATCH_CAPACITY 个 LogEntry 对象
    batch_buffer.resize(BATCH_CAPACITY);

    const size_t MAIN_QUEUE_CAPACITY = LoggerConstants::DEFAULT_MAIN_QUEUE_SIZE;

    while (m_running || m_mainQueue.size_approx() > 0 || !m_fallbackQueue.empty() || m_pendingFallbackEntry.has_value()) {
        bool did_work = false;

        // 如果有待处理的回流日志，优先处理
        if (m_pendingFallbackEntry.has_value()) {
            // 不需要拷贝，不过需要注意
            // LogEntry pendingFallbackEntry_copy = m_pendingFallbackEntry.value();
            if (m_mainQueue.try_enqueue(std::move( m_pendingFallbackEntry.value()))) {
                m_pendingFallbackEntry.reset();
                did_work = true;
            }
        }

        // 从主无锁队列出队并处理
        size_t dequeued_count = m_mainQueue.try_dequeue_bulk(batch_buffer.data(), batch_buffer.capacity()); 
        if (dequeued_count > 0) {
            did_work = true;
            std::vector<std::shared_ptr<ISink>> current_sinks;
            {
                // 这里先搞个副本，然后使用这个副本输出，这样就算进行addSink操作也不会影响这一次的输出
                // 这样缩小了锁覆盖的范围(粒度)，不用把整个for循环罩住
                std::lock_guard<std::mutex> lock(m_sinks_mutex);
                current_sinks = m_sinks;
            }
            for (size_t i = 0; i < dequeued_count; ++i) {
                auto& log_entry = batch_buffer[i];
                std::string formatted_message = m_formatter->format(log_entry);

                for (const  auto& sink : current_sinks) {
                    if (sink) {
                        sink->log(log_entry.level, formatted_message);
                    }
                }
            }
        } 

        // 如果主队列有空间且备用队列有日志，则进行回流(只有当主队列的使用率低于某个阈值时才进行回流，避免回流本身导致主队列再次满)
        // 如果没有待处理的回流日志，才去备用队列中回流 
        if (!m_pendingFallbackEntry.has_value() && !m_fallbackQueue.empty()) {
            size_t main_queue_approx_size = m_mainQueue.size_approx();

            if (main_queue_approx_size < MAIN_QUEUE_CAPACITY * m_mainQueueSpaceThreshold) {
                // 进行回流
                size_t flowed_count = drain_fallback_to_main();
                if (flowed_count > 0) {
                    did_work = true;
                }
            }
        }


        auto now = std::chrono::steady_clock::now();
        if (did_work || (now - last_flush_time) > m_flushInterval) {
            // 如果有工作或者到达时间间隔，进行一次强制刷新
            // 但是这里只是处理队列剩余，不会强制刷新底层Sink，因为还没写Sink.flush   
            // m_asyncWorker.flush;   // 感觉不太好，嗯~
            last_flush_time = now;
        }



        // 如果队列为空且没有处理任何日志，则短暂休眠，避免空转
        if (!did_work) {
            if (!m_running && m_mainQueue.size_approx() == 0 && m_fallbackQueue.empty() && !m_pendingFallbackEntry.has_value()) {
                // 如果不再工作，且队列都空了，直接退出
                break;
            }

            std::unique_lock<std::mutex> lock(m_worker_mutex);

            // 等待条件：被notify_consumer()唤醒，超时，队列已关闭
            m_worker_cv.wait_for(lock, m_flushInterval, [this]{ return !m_running || m_mainQueue.size_approx() != 0 || !m_fallbackQueue.empty(); });
        }
    }
    // 线程退出前，再次强制刷新所有剩余日志
    flush();
    std::cout << "AsyncWorker thread stopped." << std::endl;
}


size_t AsyncWorker::drain_fallback_to_main() {
    size_t re_enqueued_count = 0;
    for (size_t i = 0; i < m_fallbackBatchSize; ++i) {
        LogEntry fallback_entry;
        if (m_fallbackQueue.try_dequeue(fallback_entry)) {
            // 搞个副本，谨防移动语义陷阱 （不需要了
            // LogEntry fallback_entry_copy = fallback_entry;
            if (m_mainQueue.try_enqueue(std::move(fallback_entry))) {
                re_enqueued_count++;
            } else {
                try {
                    // m_fallbackQueue.enqueue_front(std::move(fallback_entry));
                    m_pendingFallbackEntry = std::move(fallback_entry);
                } catch (const std::runtime_error& e) {
                    std::cerr << "Error: Log fallback queue Shutting down, log droped: " << e.what() << std::endl;
                }
                break; // 停止本次回流批次
            }
        } else {
            break; // 备用队列已空
        }
    }
    return re_enqueued_count;
}
